/**
 * Created by Administrator on 2025/5/7.
 * */
#include "DataRepository.h"
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>

BaseObject *createDataRepository() {
    return new DataRepository();
}

void destroyDataRepository(BaseObject *p) {
    SAFE_DELETE(p);
}

DataRepository::DataRepository(){
    prevStep = -1;
    // 订阅数据源主题Topic
    sub = zmq::socket_t(ctx, zmq::socket_type::sub);
    sub.connect(topicAddr.c_str());
    sub.set(zmq::sockopt::subscribe, topicName);
    // AICFARNet通知通道
    req_AICFARNet_socket = zmq::socket_t(ctx, zmq::socket_type::req);
    req_AICFARNet_socket.connect(req_AICFARNet_address);
    // AIPCNet通知通道
    req_AIPCNet_socket = zmq::socket_t(ctx, zmq::socket_type::req);
    req_AIPCNet_socket.connect(req_AIPCNet_address);
    // AIE2ENet通知通道
    req_AIE2ENet_socket = zmq::socket_t(ctx, zmq::socket_type::req);
    req_AIE2ENet_socket.connect(req_AIE2ENet_address);
}

DataRepository::~DataRepository() {
}

/* 如果使用了DDS_Reader或者DDS_Writer，则对应的input或者output前64个字节为DataFormat head首部 */
void DataRepository::onCompute(buffer_table_t *input, buffer_table_t *output) {
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    zmq::message_t msgTopicName;
    zmq::message_t msgData;
    if (!sub.recv(msgTopicName, zmq::recv_flags::none)) {
        std::cerr << "Subscribe: Failed to receive topic name" << std::endl;
    }
    std::string topicName_(static_cast<char*>(msgTopicName.data()), msgTopicName.size());
    if (!sub.recv(msgData, zmq::recv_flags::none)) {
        std::cerr << "Subscribe: Failed to receive data" << std::endl;
    }
    int step = 0;
    memcpy(&step, msgData.data(), sizeof(int));
    std::cout << "DataRepository::onCompute step=" << step << "; prevStep=" << prevStep << "; STEP2=" << STEP2 << ";   ???????????" << std::endl;
    if (prevStep == step) {
        return ;
    }
    if (STEP2 == step) {
        std::cout << "DataCenter：通知AIPCTrain::AIPCNet, AICFARTrain::AICFARNet, AIE2ETrain::AIE2ENet开始训练" << std::endl;
        // 通知AIPCTrain
        int *pState = reinterpret_cast<int*>(output->list[1]->data);
        pState[0] = 1;
        // 通知AICFARTrain
        int *pAICFARNetState = reinterpret_cast<int*>(output->list[2]->data);
        pAICFARNetState[0] = 1;
        // 通知AIE2ETrain
        int *pAIE2ENetState = reinterpret_cast<int*>(output->list[3]->data);
        pAIE2ENetState[0] = 1;
    } else {
        // 恢复AIPCTrain
        // output->list[1]->len = 0;
        output->list[1]->valid = 0;
        int *pState = reinterpret_cast<int*>(output->list[1]->data);
        pState[0] = -1;
        // 恢复AICFARTrain
        output->list[2]->len = 0;
        output->list[2]->valid = 0;
        int *pAICFARNetState = reinterpret_cast<int*>(output->list[2]->data);
        pAICFARNetState[0] = 0;
        // 恢复AIE2ETrain
        output->list[3]->len = 0;
        output->list[3]->valid = 0;
        int *pAIE2ENetState = reinterpret_cast<int*>(output->list[3]->data);
        pAIE2ENetState[0] = 0;
    }
    prevStep = step;
}
